{
    "$schema": "http://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "parameters": {
        "factoryName": {
            "type": "string",
            "metadata": "Data Factory name"
        },
        "LSInsightDW_connectionString": {
            "type": "secureString",
            "metadata": "Secure string for 'connectionString' of 'LSInsightDW'"
        },
        "LSCentralOnPremDB_connectionString": {
            "type": "secureString",
            "metadata": "Secure string for 'connectionString' of 'LSCentralOnPremDB'"
        },
        "CompanyName": {
            "type" : "String"
        },
        "TablePrefix": {
            "type" : "String"
        },
        "PipelineStartDate": {
            "type": "String"
        },
        "PipelineStartTime": {
            "type": "String",
            "allowedValues": [ "01","02","03","04","05","06","07","08","09","10","11","12","13","14","15","16","17","18","19","20","21","22","23" ]
        },
        "LSInsightServerName": {
            "type": "String"
        },
        "LSInsightDatabaseName": {
            "type": "String"
        },
        "UserName": {
            "type": "String"
        },
        "Password": {
            "type": "String"
        },
        "ServerName": {
            "type": "String"
        },
        "DatabaseName": {
            "type": "String"
        },
        "SqlLogin": {
            "type": "String"
        },
        "SqlPassword": {
            "type": "String"
        }
    },
    "variables": {
        "factoryId": "[concat('Microsoft.DataFactory/factories/', parameters('factoryName'))]"
    },
    "resources": [
        {
            "name": "[concat(parameters('factoryName'), '/fItemCostAdjustment')]",
            "type": "Microsoft.DataFactory/factories/datasets",
            "apiVersion": "2018-06-01",
            "properties": {
                "linkedServiceName": {
                    "referenceName": "LSInsightDW",
                    "type": "LinkedServiceReference"
                },
                "annotations": [],
                "type": "AzureSqlTable",
                "schema": [
                    {
                        "name": "Item No_",
                        "type": "nvarchar"
                    },
                    {
                        "name": "Variant Code",
                        "type": "nvarchar"
                    },
                    {
                        "name": "Location Code",
                        "type": "nvarchar"
                    },
                    {
                        "name": "ValidFromDate",
                        "type": "datetime",
                        "precision": 23,
                        "scale": 3
                    },
                    {
                        "name": "ValidToDate",
                        "type": "datetime",
                        "precision": 23,
                        "scale": 3
                    },
                    {
                        "name": "WeightedItemAverageCost",
                        "type": "decimal",
                        "precision": 38,
                        "scale": 6
                    },
                    {
                        "name": "Batchdate",
                        "type": "datetime",
                        "precision": 23,
                        "scale": 3
                    }
                ],
                "typeProperties": {
                    "schema": "DW",
                    "table": "fItemCostAdjustment"
                }
            }
        },
        {
            "name": "[concat(parameters('factoryName'), '/PL-fItemCostAdjustment')]",
            "properties": {
                "description": "a SQL select statement to fill the table DW.fItemCostAdjustment using FULL LOAD as this captures the cost adjustments in the Value Entry Table. \n\nThis pipeline will need to run after the staging pipelines and before the fact tables, could be added to the outrigger run batch.",
                "activities": [
                    {
                        "name": "LookupLastRowVersion",
                        "description": "Retrieve  the max rowversion value stored in the control table",
                        "type": "Lookup",
                        "dependsOn": [
                            {
                                "activity": "SourceTable",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "source": {
                                "type": "AzureSqlSource",
                                "sqlReaderQuery": {
                                    "value": "@concat('SELECT max([NewRowVersion])  AS LastRowID \n  FROM ', pipeline().parameters.Control_Table_Table_Name, '\n  where [Table] =''',variables('DestinationTable'),'''')",
                                    "type": "Expression"
                                }
                            },
                            "dataset": {
                                "referenceName": "LSInsightAudit",
                                "type": "DatasetReference"
                            },
                            "firstRowOnly": true
                        }
                    },
                    {
                        "name": "LookupCurrentWaterMark",
                        "description": "Retrieve  the current maximum value in watermark column of source data store",
                        "type": "Lookup",
                        "dependsOn": [
                            {
                                "activity": "SourceTable",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "source": {
                                "type": "AzureSqlSource",
                                "sqlReaderQuery": {
                                    "value": "@concat('select COALESCE(MAX(convert(bigint, ',pipeline().parameters.Data_Source_WaterMarkColumn, ')),0) as NewWatermarkValue from ',variables('Data_Source_Table_Name'))",
                                    "type": "Expression"
                                },
                                "queryTimeout": "02:00:00"
                            },
                            "dataset": {
                                "referenceName": "LSInsightAudit",
                                "type": "DatasetReference"
                            }
                        }
                    },
                    {
                        "name": "UpdateLSInsightAudit",
                        "description": "Stored procedure activity to store the new Pipeline run metadata in LSAudit table",
                        "type": "SqlServerStoredProcedure",
                        "dependsOn": [
                            {
                                "activity": "stgCopyActivity",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "storedProcedureName": "[[dbo].[UpdateLSInsightAudit]",
                            "storedProcedureParameters": {
                                "ADFRunGUID": {
                                    "value": {
                                        "value": "@pipeline().RunId",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "CopyDuration": {
                                    "value": {
                                        "value": "@activity('stgCopyActivity').output.copyDuration",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "PipelineName": {
                                    "value": {
                                        "value": "@pipeline().Pipeline",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineStart": {
                                    "value": {
                                        "value": "@variables('PipelineStart')",
                                        "type": "Expression"
                                    },
                                    "type": "DateTime"
                                },
                                "RowsCopied": {
                                    "value": {
                                        "value": "@activity('stgCopyActivity').output.rowsCopied",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowsRead": {
                                    "value": {
                                        "value": "@activity('stgCopyActivity').output.rowsRead",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowVersion": {
                                    "value": {
                                        "value": "@activity('LookupCurrentWaterMark').output.firstRow.NewWatermarkValue",
                                        "type": "Expression"
                                    },
                                    "type": "Int64"
                                },
                                "Table": {
                                    "value": {
                                        "value": "@variables('DestinationTable')",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineTriggerType": {
                                    "value": {
                                        "value": "@pipeline().TriggerType",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "ErrorMessage": {
                                    "value": null,
                                    "type": "String"
                                },
                                "Note": {
                                    "value": "Full Load",
                                    "type": "String"
                                }
                            }
                        },
                        "linkedServiceName": {
                            "referenceName": "LSInsightDW",
                            "type": "LinkedServiceReference"
                        }
                    },
                    {
                        "name": "stgCopyActivity",
                        "type": "Copy",
                        "dependsOn": [
                            {
                                "activity": "LookupLastRowVersion",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            },
                            {
                                "activity": "LookupCurrentWaterMark",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            },
                            {
                                "activity": "PipelineStart",
                                "dependencyConditions": [
                                    "Succeeded"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30,
                            "secureOutput": false,
                            "secureInput": false
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "source": {
                                "type": "AzureSqlSource",
                                "sqlReaderQuery": {
                                    "value": "@Concat('WITH ValueEntry\nAS\n(SELECT\nvVE.[Item No_]\n,vVE.[Variant Code]\n,vVE.[Location Code]\n,vVE.[Posting Date] AS ValidFromDate\n,IIF(SUM(vVE.[SUM$Cost Amount (Actual)]) = 0 AND SUM(vVE.[SUM$Invoiced Quantity]) = 0\n, SUM(vVE.[SUM$Cost Amount (Expected)]) / NULLIF(SUM([SUM$Item Ledger Entry Quantity]), 0)\n, SUM(vVE.[SUM$Cost Amount (Actual)]) / NULLIF(SUM(vVE.[SUM$Invoiced Quantity]), 0)) AS ''WeightedItemAverageCost''\nFROM [Stg].[ItemValueEntry] vVE\nWHERE vVE.[Item Ledger Entry Type] = 1\nGROUP BY vVE.[Item No_]\n,vVE.[Posting Date]\n,vVE.[Variant Code]\n,vVE.[Location Code])\n,\ntt as (\nSELECT [Item No_]\n,[Variant Code]\n,[Location Code]\n,Min(ValidFromDate) as ValidFromDate\n,WeightedItemAverageCost from ValueEntry\ngroup by \n[Item No_]\n,[Variant Code]\n,[Location Code]\n,WeightedItemAverageCost\n)SELECT\ntt.[Item No_]\n,tt.[Variant Code]\n,tt.[Location Code]\n ,IIF(LEAD(MIN(tt.ValidFromDate)) OVER (PARTITION BY tt.[Item No_], tt.[Variant Code], tt.[Location Code] ORDER BY tt.ValidFromDate DESC) IS NULL,CAST(''1901-01-01'' AS DATE), Min(tt.ValidFromDate)) AS ValidFromDate\n,ISNULL(DATEADD(DAY, -1, LAG(MIN(tt.ValidFromDate)) OVER (PARTITION BY tt.[Item No_], tt.[Variant Code], tt.[Location Code] ORDER BY tt.ValidFromDate DESC)), CAST(GETUTCDATE() AS DATE)) AS ValidToDate\n,WeightedItemAverageCost AS WeightedItemAverageCost\n,CONVERT(DATE, GETUTCDATE()) AS ''Batchdate''\nFROM tt WHERE WeightedItemAverageCost IS NOT NULL\nGROUP BY tt.[Item No_]\n,tt.[Variant Code]\n,tt.[Location Code]\n,tt.[ValidFromDate]\n,WeightedItemAverageCost')",
                                    "type": "Expression"
                                },
                                "queryTimeout": "02:00:00"
                            },
                            "sink": {
                                "type": "AzureSqlSink",
                                "preCopyScript": {
                                    "value": "@{Concat('Delete From ', variables('DestinationTable'))}",
                                    "type": "Expression"
                                },
                                "disableMetricsCollection": false
                            },
                            "enableStaging": false
                        },
                        "inputs": [
                            {
                                "referenceName": "stgItemValueEntry",
                                "type": "DatasetReference"
                            }
                        ],
                        "outputs": [
                            {
                                "referenceName": "fItemCostAdjustment",
                                "type": "DatasetReference"
                            }
                        ]
                    },
                    {
                        "name": "PipelineStart",
                        "type": "SetVariable",
                        "dependsOn": [],
                        "userProperties": [],
                        "typeProperties": {
                            "variableName": "PipelineStart",
                            "value": {
                                "value": "@utcnow()",
                                "type": "Expression"
                            }
                        }
                    },
                    {
                        "name": "SourceTable",
                        "type": "SetVariable",
                        "dependsOn": [],
                        "userProperties": [],
                        "typeProperties": {
                            "variableName": "Data_Source_Table_Name",
                            "value": {
                                "value": "@pipeline().parameters.Data_Source_Table_Name",
                                "type": "Expression"
                            }
                        }
                    },
                    {
                        "name": "SqlServerStoredProcedure1",
                        "description": "Stored procedure activity to store the new Pipeline run metadata in LSAudit table",
                        "type": "SqlServerStoredProcedure",
                        "dependsOn": [
                            {
                                "activity": "stgCopyActivity",
                                "dependencyConditions": [
                                    "Failed"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "storedProcedureName": "[[dbo].[UpdateLSInsightAudit]",
                            "storedProcedureParameters": {
                                "ADFRunGUID": {
                                    "value": {
                                        "value": "@pipeline().RunId",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "CopyDuration": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "PipelineName": {
                                    "value": {
                                        "value": "@concat('Failed Run - ',  pipeline().Pipeline)",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineStart": {
                                    "value": {
                                        "value": "@variables('PipelineStart')",
                                        "type": "Expression"
                                    },
                                    "type": "DateTime"
                                },
                                "RowsCopied": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowsRead": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowVersion": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int64"
                                },
                                "Table": {
                                    "value": {
                                        "value": "@variables('Data_Source_Table_Name')",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineTriggerType": {
                                    "value": {
                                        "value": "@pipeline().TriggerType",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "ErrorMessage": {
                                    "value": {
                                        "value": "@activity('stgCopyActivity').Error.Message",
                                        "type": "Expression"
                                    },
                                    "type": "string"
                                }
                            }
                        },
                        "linkedServiceName": {
                            "referenceName": "LSInsightDW",
                            "type": "LinkedServiceReference"
                        }
                    },
                    {
                        "name": "SqlServerStoredProcedure2",
                        "description": "Stored procedure activity to store the new Pipeline run metadata in LSAudit table",
                        "type": "SqlServerStoredProcedure",
                        "dependsOn": [
                            {
                                "activity": "LookupCurrentWaterMark",
                                "dependencyConditions": [
                                    "Failed"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "storedProcedureName": "[[dbo].[UpdateLSInsightAudit]",
                            "storedProcedureParameters": {
                                "ADFRunGUID": {
                                    "value": {
                                        "value": "@pipeline().RunId",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "CopyDuration": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "PipelineName": {
                                    "value": {
                                        "value": "@concat('Failed Run - ',  pipeline().Pipeline)",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineStart": {
                                    "value": {
                                        "value": "@variables('PipelineStart')",
                                        "type": "Expression"
                                    },
                                    "type": "DateTime"
                                },
                                "RowsCopied": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowsRead": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowVersion": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int64"
                                },
                                "Table": {
                                    "value": {
                                        "value": "@variables('Data_Source_Table_Name')",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineTriggerType": {
                                    "value": {
                                        "value": "@pipeline().TriggerType",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "ErrorMessage": {
                                    "value": {
                                        "value": "@activity('LookupCurrentWaterMark').Error.Message",
                                        "type": "Expression"
                                    },
                                    "type": "string"
                                }
                            }
                        },
                        "linkedServiceName": {
                            "referenceName": "LSInsightDW",
                            "type": "LinkedServiceReference"
                        }
                    },
                    {
                        "name": "SqlServerStoredProcedure3",
                        "description": "Stored procedure activity to store the new Pipeline run metadata in LSAudit table",
                        "type": "SqlServerStoredProcedure",
                        "dependsOn": [
                            {
                                "activity": "LookupLastRowVersion",
                                "dependencyConditions": [
                                    "Failed"
                                ]
                            }
                        ],
                        "policy": {
                            "timeout": "7.00:00:00",
                            "retry": 0,
                            "retryIntervalInSeconds": 30
                        },
                        "userProperties": [],
                        "typeProperties": {
                            "storedProcedureName": "[[dbo].[UpdateLSInsightAudit]",
                            "storedProcedureParameters": {
                                "ADFRunGUID": {
                                    "value": {
                                        "value": "@pipeline().RunId",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "CopyDuration": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "PipelineName": {
                                    "value": {
                                        "value": "@concat('Failed Run - ',  pipeline().Pipeline)",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineStart": {
                                    "value": {
                                        "value": "@variables('PipelineStart')",
                                        "type": "Expression"
                                    },
                                    "type": "DateTime"
                                },
                                "RowsCopied": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowsRead": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int32"
                                },
                                "RowVersion": {
                                    "value": {
                                        "value": "0",
                                        "type": "Expression"
                                    },
                                    "type": "Int64"
                                },
                                "Table": {
                                    "value": {
                                        "value": "@variables('Data_Source_Table_Name')",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "PipelineTriggerType": {
                                    "value": {
                                        "value": "@pipeline().TriggerType",
                                        "type": "Expression"
                                    },
                                    "type": "String"
                                },
                                "ErrorMessage": {
                                    "value": {
                                        "value": "@activity('LookupLastRowVersion').Error.Message",
                                        "type": "Expression"
                                    },
                                    "type": "string"
                                }
                            }
                        },
                        "linkedServiceName": {
                            "referenceName": "LSInsightDW",
                            "type": "LinkedServiceReference"
                        }
                    }
                ],
                "parameters": {
                    "Data_Source_Table_Name": {
                        "type": "String",
                        "defaultValue": "[[Stg].[ItemValueEntry]"
                    },
                    "Data_Source_WaterMarkColumn": {
                        "type": "String",
                        "defaultValue": "[[RowID]"
                    },
                    "Control_Table_Table_Name": {
                        "type": "String",
                        "defaultValue": "[[dbo].[LSInsightAudit]"
                    },
                    "Control_Table_Column_Name": {
                        "type": "String",
                        "defaultValue": "[[NewRowVersion]"
                    },
                    "TablePrefix": {
                        "type": "string",
                        "defaultValue": "[parameters('TablePrefix')]"
                    }
                },
                "variables": {
                    "PipelineStart": {
                        "type": "String"
                    },
                    "DestinationTable": {
                        "type": "String",
                        "defaultValue": "[[DW].[fItemCostAdjustment]"
                    },
                    "Data_Source_Table_Name": {
                        "type": "String"
                    },
                    "ErrorMessage": {
                        "type": "String"
                    }
                },
                "folder": {
                    "name": "FactTables"
                },
                "annotations": []
            },
            "type": "Microsoft.DataFactory/factories/pipelines",
            "apiVersion": "2018-06-01",
            "dependsOn": [
                "[concat(variables('factoryId'), '/datasets/fItemCostAdjustment')]"
            ]
        }
    ]
}